Starting this PR in draft so we can review early and gather alignment.

Summary

The solution is mostly based on discussion and proposals in #23 :

• Target driven: the user sends CLUSTER IMPORT to the target and it syncs down the slot
• Uses AOF format for snapshot, proxying all commands directly from source primary through the target primary to all target replicas
• Consensus-less epoch bump to broadcast the topology change

Import and Export Workflows

We utilize a new internal CLUSTER SYNCSLOTS command with additional sub-commands to transition through the slot migration state machine.

On either end of the migration, we track the ongoing migrations via two job queues: slot_import_jobs and slot_export_jobs. Right now, we only support one concurrent job in both the slot import and slot export job queue. There is no design restriction here (outside of perhaps some protocol additions to CLUSTER SYNCSLOTS - but it helps with simplicity for the first iteration.

Workflow overview

1. User sends CLUSTER IMPORT SLOTSRANGE <slot start> <slot end> ... to target node T
2. T initiates a new connection to source node S
3. If required, T runs AUTH based on replication configuration
4. T initiates CLUSTER SYNCSLOTS START to S
5. S begins tracking the client that sent the command as the slot export client. It spawns a child process at the next available time and runs AOF rewrite with just the specified slots. It then begins accumulating a backlog of writes in the slot export client output buffer, without installing the write handler.
6. At the end of the AOF rewrite, S also appends CLUSTER SYNCSLOTS ENDSNAPSHOT
7. T processes the AOF rewrite as it would any other client using readQueryFromClient. Once it gets the CLUSTER SYNCSLOTS ENDSNAPSHOT, T sends back a CLUSTER SYNCSLOTS PAUSE to pause S.
8. Upon getting the command, S unblocks the slot export client to T which has been accumulating ongoing writes. S then pauses itself, and sends CLUSTER SYNCSLOTS PAUSEOFFSET <offset> back to T with the current offset. (Note that the offset is not the primary replication offset, it is actually a computed offset based on how much we have been accumulating on T's client.)
9. T waits for it's replication offset to catch up to the sent offset, and once caught up executes the consensus-less bump.
10. S finds out about the bump via cluster gossip, unpauses itself, and cleans up dirty keys.

If at any point a client is disconnected on either end, or a timeout is reached on the target node, the migration is marked as failed. If a migration fails, we delete all keys in the slots we were migrating.

Filtering traffic

We filter the traffic to the target node through a filtered AOF rewrite and a filtered replication stream. The filtered AOF rewrite requires some refactoring of the snapshot code for reuse, but utilizes the same overall procedure (piping through the parent process to the target node connection).

The filtered replication stream hooks into the existing replication code and appends the commands directly to the client output buffer. We don't use replication backlog as there is no easy way to filter it once added to the backlog without re-processing it to query for the slot number of each command.

We add a new check in putClientInPendingWriteQueue to prevent the two command streams from merging. The parent process will just accumulate the replication stream in the client output buffer until we get the notification that the target is done with the snapshot.

Other notes

• Since there are many checks that apply to both slot migration and replication - I added a new client flag "replicated" which simply means that a client is being replicated from another node. In many places, we now check the replicated flag instead of the primary flag.
• To support non-contiguous slot ranges, the APIs and code use a slot bitmap to share information about what slots are being migrated.
• We already delete unowned slots after loading RDB and AOF. Replicas that are promoted to primary now go through the same steps to delete all unowned keys. This prevents the keys from being leaked forever if the previous primary disappears.

Remaining work items

• Continue cleaning up code, especially around the state machines
• TCL tests
• Metrics
• Improved logging
• Improved pause logic - where we only pause the slots being moved
• Hide the in-progress import better for commands like KEYS and RANDOMKEY
• Other management APIs (i.e. CLUSTER IMPORT STATUS, CLUSTER IMPORT CANCEL)
• Incorporate dual-channel-style replication